package com.lanou.quanttradedata.stock.mapreduce.reducer;

import com.lanou.quanttradedata.QuantTradeDataApplication;
import com.lanou.quanttradedata.stock.javabean.MRDailyResult;
import com.lanou.quanttradedata.stock.mapper.MapReduceDailyResultMapper;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * 这里是业务层的减量器，
 * 可以对数据进行业务筛选计算，在映射类处理后，将映射类映射合并的数据进行进一步业务计算，
 * 在此计算出最终结果，保存到HDFS中，并将结果保存数据库
 */
public class StockDailyInfoAvgReducer extends Reducer<Text,Text,Text,Text> {
    //mapper层注入
    private MapReduceDailyResultMapper mapReduceDailyResultMapper = (MapReduceDailyResultMapper) QuantTradeDataApplication.context.getBean("mapReduceDailyResultMapper");
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //昨日收盘价
        float pric = 0;
        //涨跌幅
        float pct = 0;
        //成交额
        float amount = 0;
        //统计总共有几天的数据
        int count = 0;
        //累加计算
        for (Text value : values) {
            String[] temps = value.toString().split(",");
            pric += Double.parseDouble(temps[6]);
            pct += Double.parseDouble(temps[8]);
            amount += Double.parseDouble(temps[10]);
            count++;
        }
        //求count天的平均
        pric = pric/count;
        pct = pct/count;
        amount = amount/count;
        //结果写入到数据库中
        mapReduceDailyResultMapper.insertMRDailyResult(new MRDailyResult(null,key.toString().
                replace("\"", ""),pric,pct,amount));
        //结果写出到hdfs中
        context.write(key, new Text(pric + " " + pct+ " " + amount));
    }
}
